import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import uuid
from fastapi.responses import HTMLResponse
from py_db import cur
import sqlite3

app = FastAPI()

ws_list = []


# 保存目前有页面的客户端


def return_ok(data=None):
    return {
        "code": 200,
        "data": data,
        "msg": ""
    }


def return_error(msg=None):
    default_msg = "请求错误"
    if msg:
        default_msg = msg
    return {
        "code": 500,
        "msg": default_msg
    }


@app.get("/")
def indenx():
    global User_ID
    html_content = ""
    with open("web_kafka.html", 'r') as f:
        html_content = f.read()
    # _id = str(uuid.uuid4())
    # User_ID[_id] = init_user_db()
    return HTMLResponse(content=html_content, status_code=200)


@app.get("/topics")
async def get_topic():
    con = sqlite3.connect("webkafka.db")
    cur = con.cursor()
    db_obj = cur.execute("select topic_name, status, id from topic")
    datas = db_obj.fetchall()
    r_data = []
    print(datas, "all_data")
    for data in datas:
        r_data.append({"topic_name": data[0], "status": data[1], "id": data[2]})
    cur.close()
    con.close()
    return return_ok(r_data)


from fastapi import Body


@app.post("/topic")
async def post_topic(request=Body([])):
    print("in")
    con = sqlite3.connect("webkafka.db")
    cur = con.cursor()
    # print(request)
    # topic_datas = request.get("topic_datas")
    clear_data = []
    topic_datas = request.get("datas", [])
    for data in topic_datas:
        clear_data.append((str(uuid.uuid4()), data, 0))
    cur.executemany("INSERT INTO topic(id, topic_name, status) VALUES(?,?,?)", clear_data)
    con.commit()
    cur.close()
    con.close()
    return return_ok()


@app.put("/update_topic")
async def post_topic(request=Body([])):
    con = sqlite3.connect("webkafka.db")
    cur = con.cursor()
    topic_name = request.get("topic_name")
    status = request.get("status")
    cur.execute("UPDATE topic SET status = ? where topic_name = ?", (status, topic_name))
    con.commit()
    cur.close()
    con.close()
    return return_ok()


class ConnectionManager:
    def __init__(self):
        # 存放激活的ws连接对象
        self.active_connections = []

    async def connect(self, ws: WebSocket):
        # 等待连接
        await ws.accept()
        # 存储ws连接对象
        self.active_connections.append(ws)

    def disconnect(self, ws: WebSocket):
        # 关闭时 移除ws对象
        self.active_connections.remove(ws)

    @staticmethod
    async def send_personal_message(message: str, ws: WebSocket):
        # 发送个人消息
        await ws.send_text(message)

    async def broadcast(self, message: str):
        # 广播消息
        for connection in self.active_connections:
            await connection.send_text(message)


import time
import threading

#
# async def send_kafka_data(websocket):
#     try:
#         while 1:
#             await manager.connect(websocket)
#             await manager.send_personal_message(f"你说了: aaaa", websocket)
#             time.sleep(2)
#     except:
#         print("websocket exit")
#
#
# @app.websocket("/ws")
# def websocket_endpoint(websocket: WebSocket):
#     t1 = threading.Thread(target=send_kafka_data, args=(websocket,))
#     t1.start()

import asyncio


@app.websocket("/ws/{user}")
async def websocket_endpoint(websocket: WebSocket, user: str):
    manager = ConnectionManager()

    await manager.connect(websocket)

    await manager.broadcast(f"用户{user}进入聊天室")
    data = '0000'
    # await manager.send_personal_message(f"你说了: {data}", websocket)

    try:
        i = 0
        while True:
            # data = await websocket.receive_text()
            # await manager.send_personal_message(f"你说了: {data}", websocket)
            # await manager.broadcast(f"用户:{user} 说: {data}")
            await manager.send_personal_message("发送数据{}".format(str(i)) * 88, websocket)
            await asyncio.sleep(3)
            i += 1

    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"用户-{user}-离开")


# if __name__ == "__main__":
#     print(load_db())
#     db_json = {"topic":{}}
#     #save_db(db_json)

if __name__ == "__main__":
    import uvicorn

    # 官方推荐是用命令后启动 uvicorn main:app --host=127.0.0.1 --port=8010 --reload
    uvicorn.run(app='web_kafka:app', host="127.0.0.1", port=8080, reload=True, debug=True)
